package cn.sheep.dmp.etl

import cn.sheep.dmp.utils.Tools
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types._
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 原始日志转换成parquet文件
  * Sheep.Old @ 64341393
  * Created 2018/3/28
  */
object Biz2Parquet {

    def main(args: Array[String]): Unit = {

        // 接受参数
        if (args.length != 3) {
            println(
                """
                  |cn.sheep.dmp.etl.Biz2Parquet
                  |参数：
                  |     dataInputPath <原始日志所在路径>
                  |     compressionCode <压缩格式>
                  |     outputPath  <转换后的日志存储路径>
                """.stripMargin)
            sys.exit()
        }

        val Array(dataInputPath, compressionCode,outputPath) = args



        val sparkConf = new SparkConf().setAppName("日志转parquet文件")
          .setMaster("local[*]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(sparkConf)


        // 读取数据
        val filtered: RDD[Array[String]] = sc.textFile(dataInputPath).map(_.split(",", -1)).filter(_.length >= 85)

        // 日志转parquet <= schema + RDD[Row]
        val row: RDD[Row] = filtered.map(arr => {
            Row(
              arr(0),
              Tools.str2Int(arr(1)),
              Tools.str2Int(arr(2)),
              Tools.str2Int(arr(3)),
              Tools.str2Int(arr(4)),
              arr(5),
              arr(6),
              Tools.str2Int(arr(7)),
              Tools.str2Int(arr(8)),
              Tools.str2Double(arr(9)),
              Tools.str2Double(arr(10)),
              arr(11),
              arr(12),
              arr(13),
              arr(14),
              arr(15),
              arr(16),
              Tools.str2Int(arr(17)),
              arr(18),
              arr(19),
              Tools.str2Int(arr(20)),
              Tools.str2Int(arr(21)),
              arr(22),
              arr(23),
              arr(24),
              arr(25),
              Tools.str2Int(arr(26)),
              arr(27),
              Tools.str2Int(arr(28)),
              arr(29),
              Tools.str2Int(arr(30)),
              Tools.str2Int(arr(31)),
              Tools.str2Int(arr(32)),
              arr(33),
              Tools.str2Int(arr(34)),
              Tools.str2Int(arr(35)),
              Tools.str2Int(arr(36)),
              arr(37),
              Tools.str2Int(arr(38)),
              Tools.str2Int(arr(39)),
              Tools.str2Double(arr(40)),
              Tools.str2Double(arr(41)),
              Tools.str2Int(arr(42)),
              arr(43),
              Tools.str2Double(arr(44)),
              Tools.str2Double(arr(45)),
              arr(46),
              arr(47),
              arr(48),
              arr(49),
              arr(50),
              arr(51),
              arr(52),
              arr(53),
              arr(54),
              arr(55),
              arr(56),
              Tools.str2Int(arr(57)),
              Tools.str2Double(arr(58)),
              Tools.str2Int(arr(59)),
              Tools.str2Int(arr(60)),
              arr(61),
              arr(62),
              arr(63),
              arr(64),
              arr(65),
              arr(66),
              arr(67),
              arr(68),
              arr(69),
              arr(70),
              arr(71),
              arr(72),
              Tools.str2Int(arr(73)),
              Tools.str2Double(arr(74)),
              Tools.str2Double(arr(75)),
              Tools.str2Double(arr(76)),
              Tools.str2Double(arr(77)),
              Tools.str2Double(arr(78)),
              arr(79),
              arr(80),
              arr(81),
              arr(82),
              arr(83),
              Tools.str2Int(arr(84))
            )
        })


        // 构建schema
        val schema = StructType(
            Seq(
                StructField("sessionid", StringType),
                StructField("advertisersid", IntegerType),
                StructField("adorderid", IntegerType),
                StructField("adcreativeid", IntegerType),
                StructField("adplatformproviderid", IntegerType),
                StructField("sdkversion", StringType),
                StructField("adplatformkey", StringType),
                StructField("putinmodeltype", IntegerType),
                StructField("requestmode", IntegerType),
                StructField("adprice", DoubleType),
                StructField("adppprice", DoubleType),
                StructField("requestdate", StringType),
                StructField("ip", StringType),
                StructField("appid", StringType),
                StructField("appname", StringType),
                StructField("uuid", StringType),
                StructField("device", StringType),
                StructField("client", IntegerType),
                StructField("osversion", StringType),
                StructField("density", StringType),
                StructField("pw", IntegerType),
                StructField("ph", IntegerType),
                StructField("long", StringType),
                StructField("lat", StringType),
                StructField("provincename", StringType),
                StructField("cityname", StringType),
                StructField("ispid", IntegerType),
                StructField("ispname", StringType),
                StructField("networkmannerid", IntegerType),
                StructField("networkmannername", StringType),
                StructField("iseffective", IntegerType),
                StructField("isbilling", IntegerType),
                StructField("adspacetype", IntegerType),
                StructField("adspacetypename", StringType),
                StructField("devicetype", IntegerType),
                StructField("processnode", IntegerType),
                StructField("apptype", IntegerType),
                StructField("district", StringType),
                StructField("paymode", IntegerType),
                StructField("isbid", IntegerType),
                StructField("bidprice", DoubleType),
                StructField("winprice", DoubleType),
                StructField("iswin", IntegerType),
                StructField("cur", StringType),
                StructField("rate", DoubleType),
                StructField("cnywinprice", DoubleType),
                StructField("imei", StringType),
                StructField("mac", StringType),
                StructField("idfa", StringType),
                StructField("openudid", StringType),
                StructField("androidid", StringType),
                StructField("rtbprovince", StringType),
                StructField("rtbcity", StringType),
                StructField("rtbdistrict", StringType),
                StructField("rtbstreet", StringType),
                StructField("storeurl", StringType),
                StructField("realip", StringType),
                StructField("isqualityapp", IntegerType),
                StructField("bidfloor", DoubleType),
                StructField("aw", IntegerType),
                StructField("ah", IntegerType),
                StructField("imeimd5", StringType),
                StructField("macmd5", StringType),
                StructField("idfamd5", StringType),
                StructField("openudidmd5", StringType),
                StructField("androididmd5", StringType),
                StructField("imeisha1", StringType),
                StructField("macsha1", StringType),
                StructField("idfasha1", StringType),
                StructField("openudidsha1", StringType),
                StructField("androididsha1", StringType),
                StructField("uuidunknow", StringType),
                StructField("userid", StringType),
                StructField("iptype", IntegerType),
                StructField("initbidprice", DoubleType),
                StructField("adpayment", DoubleType),
                StructField("agentrate", DoubleType),
                StructField("lomarkrate", DoubleType),
                StructField("adxrate", DoubleType),
                StructField("title", StringType),
                StructField("keywords", StringType),
                StructField("tagid", StringType),
                StructField("callbackdate", StringType),
                StructField("channelid", StringType),
                StructField("mediatype", IntegerType)
            )
        )

        val sqlc = new SQLContext(sc)
        sqlc.setConf("spark.sql.parquet.compression.codec", compressionCode)
        val dataFrame = sqlc.createDataFrame(row, schema)

        // 将dataFrame转存parquet
        dataFrame.write.parquet(outputPath)

        sc.stop()

    }

}
